package com.xxx.rpc.registry.zookeeper;

import com.xxx.rpc.registry.ServiceRegistry;

import java.util.HashMap;
import java.util.Map;

import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 基于 ZooKeeper 的服务注册接口实现
 *
 * @author huangyong & Rosun
 * @since 1.0.0
 */
public class ZooKeeperServiceRegistry implements ServiceRegistry {

    private static final Logger LOGGER = LoggerFactory.getLogger(ZooKeeperServiceRegistry.class);

    private final ZkClient zkClient;
    
    /**
     * key String serviceName, <br> 
     * value String serviceAddress<br>
     */
    private Map<String, String> regMap = new HashMap<String, String>();

    public ZooKeeperServiceRegistry(String zkAddress) {
    	LOGGER.info("#### using zookeeper url:"+ zkAddress+"\n\n");
        // 创建 ZooKeeper 客户端
        zkClient = new ZkClient(zkAddress, Constant.ZK_SESSION_TIMEOUT, Constant.ZK_CONNECTION_TIMEOUT);
        LOGGER.debug("connect zookeeper");
        zkClient.subscribeStateChanges(new IZkStateListener() {
			
			@Override
			public void handleStateChanged(KeeperState state) throws Exception {
				LOGGER.warn( "handleStateChanged,state:" + state);
			}
			
			@Override
			public void handleSessionEstablishmentError(Throwable error) throws Exception {
				
			}
			
			@Override
			public void handleNewSession() throws Exception {
				//重新注册数据
				for(String serviceName : ZooKeeperServiceRegistry.this.regMap.keySet()){
					doRegist(serviceName,ZooKeeperServiceRegistry.this.regMap.get(serviceName));
				}
			}
		});
    }

    
   
    
    @Override
    public void register(String serviceName, String serviceAddress) {
    	LOGGER.debug("begin register:{}-->{}",serviceName,serviceAddress);
    	this.regMap.put(serviceName, serviceAddress);
        doRegist(serviceName,serviceAddress);
    }




	private void doRegist(String serviceName, String serviceAddress) {
		// 创建 registry 节点（持久）
        String registryPath = Constant.ZK_REGISTRY_PATH;
        if (!zkClient.exists(registryPath)) {
            zkClient.createPersistent(registryPath);
            LOGGER.debug("create registry node: {}", registryPath);
        }
        // 创建 service 节点（持久）
        String servicePath = registryPath + "/" + serviceName;
        if (!zkClient.exists(servicePath)) {
            zkClient.createPersistent(servicePath);
            LOGGER.debug("create service node: {}", servicePath);
        }
        // 创建 address 节点（临时）
        String addressPath = servicePath + "/address-";
        String addressNode = zkClient.createEphemeralSequential(addressPath, serviceAddress);
        LOGGER.info("end doRegist :addressPath="+addressPath+",addressNode="+addressNode);
	}
}